package org.myfram.flink.flinkonjar.application.job;

import org.apache.flink.connector.hbase.sink.HBaseSinkFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.myfram.flink.flinkonjar.common.config.KafkaConfig;
import org.myfram.flink.flinkonjar.common.connector.hbase.sink.HBaseSinkBuilder;
import org.myfram.flink.flinkonjar.common.connector.kafka.sink.KafkaSinkBuilder;
import org.myfram.flink.flinkonjar.common.connector.kafka.source.KafkaSourceBuilder;
import org.myfram.flink.flinkonjar.common.properties.HBaseProperties;
import org.myfram.flink.flinkonjar.common.properties.KafkaProperties;
import org.myfram.flink.flinkonjar.common.util.YamlUtils;
import org.myfram.flink.flinkonjar.domain.entity.HBaseUserTable;
import org.myfram.flink.flinkonjar.domain.entity.KafkaTargetUserEvent;
import org.myfram.flink.flinkonjar.domain.process.AgeFilterFunction;
import org.myfram.flink.flinkonjar.domain.process.UserFieldProcessFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class KafkaToHBaseJob extends BaseFlinkJob{

    Logger log = LoggerFactory.getLogger(KafkaToHBaseJob.class.getName());

    @Override
    protected void process(StreamExecutionEnvironment env) throws Exception {
        KafkaProperties kafkaProperties = YamlUtils.getObject("flink.kafka", KafkaProperties.class);
        HBaseProperties hBaseProperties = YamlUtils.getObject("flink.hbase", HBaseProperties.class);
        Properties sourceProps = KafkaConfig.getSourceProps(kafkaProperties.getSource());
        Properties sinkProps = KafkaConfig.getSinkProps(kafkaProperties.getSink());

        DataStreamSource<String> kafkaSource = KafkaSourceBuilder.build(env, kafkaProperties.getSource().getTopic(), sourceProps);
        SingleOutputStreamOperator<KafkaTargetUserEvent> userStream = kafkaSource.map(new UserFieldProcessFunction())
                .filter(new AgeFilterFunction())
                .map(u -> {
                    log.info("源端数据：{}", u.toString());
                    return u;
                });
        HBaseSinkFunction<String> hbaseSink = new HBaseSinkBuilder<String>(new HBaseUserTable(), hBaseProperties).build();
        userStream.map(KafkaSinkBuilder::serialize)
                .addSink(hbaseSink);
    }
}
